from es_pandas import es_pandas as origin_es_pandas
from redis import ConnectionPool, Redis
from kafka import KafkaProducer

from air_web.config.config import config
from air_web.data_platform.sql_connection import MysqlPool
from air_web.data_platform.sql_process import MySQLPandas


class es_pandas(origin_es_pandas):
    def __init__(self, *args, **kwargs):
        kwargs.update(
            {"http_auth": (config.get("ES_USER"), config.get("ES_PASSWORD"))}
        )
        super().__init__(*args, **kwargs)


def init_db(mysql_db_name=config["MYSQL_DB_NAME"]):
    """初始化数据库"""
    sql_config = {
        "MYSQL_USER": config["MYSQL_USER"],
        "MYSQL_HOST": config["MYSQL_HOST"],
        "MYSQL_PORT": config["MYSQL_PORT"],
        "MYSQL_DB_NAME": mysql_db_name,
        "MYSQL_PASSWORD": config["MYSQL_PASSWORD"],
        "MYSQL_ENCODING": config["MYSQL_ENCODING"],
    }
    sql_engine = MySQLPandas(**sql_config)
    return sql_engine


sql_engine = init_db()
mysql_con = MysqlPool(
    config["MYSQL_HOST"],
    int(config["MYSQL_PORT"]),
    config["MYSQL_USER"],
    config["MYSQL_PASSWORD"],
    config["MYSQL_DB_NAME"],
)
redis_con = Redis(
    connection_pool=ConnectionPool(
        host=config["REDIS_HOST"],
        port=config["REDIS_PORT"],
        password=config["REDIS_PASSWORD"],
        db=config["REDIS_DB"],
        max_connections=30,
    )
)
